package com.atguigu.api

import java.util

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests


/**
 * @description: xxx
 * @time: 2020/6/21 13:10
 * @author: baojinlong
 **/
object ElasticSearchSinkTest {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")

    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })

    // 定义一个httpHosts
    val httpHosts = new util.ArrayList[HttpHost]
    httpHosts.add(new HttpHost("localhost", 9200))
    // 定义一个ElasticSearchSinkFunction
    val esSinkFunction: ElasticsearchSinkFunction[SensorReading] = new ElasticsearchSinkFunction[SensorReading] {
      override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
        // 包装写入es数据
        val dataSource = new util.HashMap[String, String]
        dataSource.put("sensor_id", t.id)
        dataSource.put("temp", t.temperature.toString)
        dataSource.put("ts", t.timestamp.toString)
        // 创建一个人indexRequest
        val indexRequest: IndexRequest = Requests.indexRequest()
          .index("sensor_temp_index")
          .`type`("doc")
          .source(dataSource)
        // 用indexer发送http请求
        requestIndexer.add(indexRequest)
      }
    }
    // 写入到elasticSearch
    dataStream.addSink(new ElasticsearchSink.Builder[SensorReading](httpHosts, esSinkFunction).build())

    environment.execute("sink simple test job")
  }
}
